package com.qyer.search.storm.util;

import com.qyer.search.storm.logging.InfoLogger;
import kafka.consumer.ConsumerThreadId;
import kafka.javaapi.consumer.*;

import java.util.Collections;
import java.util.Map;
import java.util.Set;

/**
 * Created by tangzheng on 16/2/2.
 */
public class KafkaListener implements ConsumerRebalanceListener {
  private InfoLogger infoLogger;
  public KafkaListener(InfoLogger infoLogger){
    this.infoLogger = infoLogger;
  }

  public void beforeReleasingPartitions(Map<String, Set<Integer>> partitionOwnership){
    if(partitionOwnership.isEmpty()){
      infoLogger.info("partitionOwnership is empty.");
    }else {
      infoLogger.info("partitionOwnership:{}", partitionOwnership);
    }
  }
  public void beforeStartingFetchers(String consumerId, Map<String, Map<Integer, ConsumerThreadId>> globalPartitionAssignment){
    infoLogger.info("consumerId:{},PartitionAssignment:{}", consumerId, globalPartitionAssignment);
  }
}
